package com.tobby.camelinaction.scalability;

import java.util.concurrent.ExecutorService;

import org.apache.camel.AsyncCallback;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.impl.DefaultAsyncProducer;

public class ErpProducer extends DefaultAsyncProducer {

	// use a thread pool for async communication with ERP
	private ExecutorService executor;

	public ErpProducer(Endpoint endpoint) {
		super(endpoint);
		// use Camel to create the thread pool for us
		this.executor = endpoint.getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, "ERP", 10);
	}

	public boolean process(Exchange exchange, final AsyncCallback callback) {
		// sumulate async communication using a thread pool in which will return a reply in 5 seconds
		executor.submit(new ErpTask(exchange, callback));

		// return false to tell Camel that we process asynchronously
		// which enables the Camel routing engine to know this and act accordingly
		// notice the ErpTask must invoke the callback.done(false) because what we return
		// here must match the boolean in the callback.done method.
		return false;
	}

	private class ErpTask implements Runnable {

		private final Exchange exchange;

		private final AsyncCallback callback;

		public ErpTask(Exchange exchange, AsyncCallback callback) {
			this.exchange = exchange;
			this.callback = callback;
		}

		public void run() {
			log.info("Calling ERP");
			// simulate the communication with ERP takes 5 seconds
			try {
				Thread.sleep(5000);
			}
			catch (InterruptedException e) {
				// ignore
			}

			log.info("ERP reply received");

			// set reply
			String in = exchange.getIn().getBody(String.class);
			exchange.getOut().setBody(in + ";516");

			// notify callback we are done
			// we must use done(false) because the process method returned false
			log.info("Continue routing");
			callback.done(false);
		}
	}

}
